package org.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.function.DimMapFunction;
import org.atguigu.gmall.realtime.util.AtguiguUtil;

import java.math.BigDecimal;
import java.time.Duration;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/5/4 20:24:35
 * Description: Todo
 * Version：1.0
 */
public class Dws_09_DwsTradeSkuOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow()
                .init(40009,2,
                        Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,"Dws_09_DwsTradeSkuOrderWindow");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // parse data to pojo
        SingleOutputStreamOperator<TradeSkuOrderBean> parseStream = parseToPojo(ds);
        // remove duplicate data according to the order details id
        SingleOutputStreamOperator<TradeSkuOrderBean> distinctedStream = distinctByOrderDetailId(parseStream);
        // window and aggregation
        SingleOutputStreamOperator<TradeSkuOrderBean> aggStream = windowAndAgg(distinctedStream);
        // supplementary dimension
        SingleOutputStreamOperator<TradeSkuOrderBean> streamWithDims = joinDims(aggStream);
        streamWithDims.print();
        //

    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> joinDims(SingleOutputStreamOperator<TradeSkuOrderBean> aggStream) {
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoStream = aggStream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_sku_info";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getSkuId();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setSkuName(dim.getString("SKU_NAME"));
                bean.setSpuId(dim.getString("SPU_ID"));
                bean.setTrademarkId(dim.getString("TM_ID"));
                bean.setCategory3Id(dim.getString("CATEGORY3_ID"));
            }
        });
        SingleOutputStreamOperator<TradeSkuOrderBean> spuInfoStream = skuInfoStream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_spu_info";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getSpuId();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setSpuName(dim.getString("SPU_NAME"));
            }
        });
        SingleOutputStreamOperator<TradeSkuOrderBean> tmStream = spuInfoStream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_base_trademark";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getTrademarkId();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setTrademarkName(dim.getString("TM_NAME"));
            }
        });
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = tmStream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_base_category3";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getCategory3Id();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setCategory3Name(dim.getString("NAME"));
                bean.setCategory2Id(dim.getString("CATEGORY2_ID"));
            }
        });
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = c3Stream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_base_category2";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getCategory2Id();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setCategory2Name(dim.getString("NAME"));
                bean.setCategory1Id(dim.getString("CATEGORY1_ID"));
            }
        });
        return c2Stream.map(new DimMapFunction<TradeSkuOrderBean>() {
            @Override
            public String getTable() {
                return "dim_base_category1";
            }

            @Override
            public String getId(TradeSkuOrderBean bean) {
                return bean.getCategory1Id();
            }

            @Override
            public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                bean.setCategory1Name(dim.getString("NAME"));
            }
        });

    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(SingleOutputStreamOperator<TradeSkuOrderBean> distinctedStream) {
        return distinctedStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((obj,ts) -> obj.getTs()*1000)
                                .withIdleness(Duration.ofSeconds(60)))
                .keyBy(TradeSkuOrderBean::getSkuId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                            @Override
                            public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                                value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                                value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                                value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                                value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String s, ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>.Context context, Iterable<TradeSkuOrderBean> elements, Collector<TradeSkuOrderBean> out) throws Exception {
                                TradeSkuOrderBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());
                                out.collect(bean);
                            }
                        });
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> distinctByOrderDetailId(SingleOutputStreamOperator<TradeSkuOrderBean> parseStream) {
        return parseStream.keyBy(TradeSkuOrderBean::getOrderDetailId)
                .process(new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private ValueState<TradeSkuOrderBean> lastBeanState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastBeanState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>(
                                        "lastBean",
                                        TradeSkuOrderBean.class)
                        );
                    }

                    @Override
                    public void processElement(TradeSkuOrderBean value, KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>.Context ctx, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean lastBean = lastBeanState.value();
                        if (lastBean == null){
                            out.collect(value);
                        }else {
                            lastBean.setOriginalAmount(value.getOriginalAmount().subtract(lastBean.getOriginalAmount()));
                            lastBean.setOrderAmount(value.getOrderAmount().subtract(lastBean.getOrderAmount()));
                            lastBean.setActivityAmount(value.getActivityAmount().subtract(lastBean.getActivityAmount()));
                            lastBean.setCouponAmount(value.getCouponAmount().subtract(lastBean.getCouponAmount()));
                            out.collect(lastBean);
                        }
                        lastBeanState.update(value);
                    }
                });
    }

    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPojo(DataStreamSource<String> ds) {
        return ds.map(new MapFunction<String, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                return TradeSkuOrderBean.builder()
                        .skuId(obj.getString("sku_id"))
                        .originalAmount(obj.getBigDecimal("split_original_amount"))
                        .activityAmount(obj.getBigDecimal("split_activity_amount")==null ? new BigDecimal("0.0") : obj.getBigDecimal("split_activity_amount"))
                        .couponAmount(obj.getBigDecimal("split_coupon_amount")==null ? new BigDecimal("0.0") : obj.getBigDecimal("split_coupon_amount"))
                        .orderAmount(obj.getBigDecimal("split_total_amount"))
                        .ts(obj.getLong("ts"))
                        .orderDetailId(obj.getString("id"))
                        .build();
            }
        });
    }
}
